package com.shenjies88.practice.reactor.thread;

import com.shenjies88.practice.reactor.hold.WorkWrapperHold;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * @author shenjies88
 * @since 2022/12/7/21:17
 */
public class SelectorThread extends Thread {

    private Selector selector;
    private final int index;
    private final LinkedBlockingQueue<SocketChannel> queue;

    public SelectorThread(LinkedBlockingQueue<SocketChannel> queue, int index) {
        this.queue = queue;
        super.setName("selector-thread-" + index);
        this.index = index;
    }

    @Override
    public void run() {
        try {
            selector = Selector.open();
        } catch (IOException e) {
            e.printStackTrace();
            System.err.println("启动selector线程异常");
            throw new RuntimeException(e);
        }
        startRegistryChannelThread();
        while (true) {
            try {
                if (selector.select(1000) == 0) {
                    continue;
                }
            } catch (IOException e) {
                e.printStackTrace();
                System.err.println("监听selector异常");
                throw new RuntimeException(e);
            }
            var selectionKeys = selector.selectedKeys();
            var iterator = selectionKeys.iterator();
            while (iterator.hasNext()) {
                var selectionKey = iterator.next();
                if (selectionKey.isReadable()) {
                    WorkWrapperHold.submit(selectionKey);
                }
                if (selectionKey.isWritable()) {
                    //TODO 写请求
                }
                iterator.remove();
            }
        }
    }

    private void startRegistryChannelThread() {
        new Thread(() -> {
            while (true) {
                SocketChannel socketChannel = null;
                try {
                    socketChannel = queue.take();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    System.err.println("注册出队异常");
                }
                if (socketChannel != null) {
                    try {
                        socketChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE, ByteBuffer.allocate(8192));
                    } catch (ClosedChannelException e) {
                        e.printStackTrace();
                        System.err.println("注册读写异常");
                    }
                }

            }
        }, "selector-take-" + index).start();
    }
}
